package com.example.study.commen.flinkcdc;


import com.alibaba.fastjson.JSONObject;
import com.example.study.config.redis.RedisUtil;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;


/**
 * @author 海绵先生
 * @Description TODO 关于DataStream方式的CDC，自定义反序列化
 * @date 2023/1/13-21:26
 */
/*官方默认String类型的数据样式
SourceRecord{sourcePartition={server=mysql_binlog_source}, sourceOffset={transaction_id=null, ts_sec=1673531342, file=mysql-bin.000001, pos=803, row=1, server_id=1, event=2}} ConnectRecord{topic='mysql_binlog_source.cdc_test.user_info', kafkaPartition=null, key=Struct{id=1003}, keySchema=Schema{mysql_binlog_source.cdc_test.user_info.Key:STRUCT}, value=Struct{before=Struct{id=1003,name=wangwu,sex=famale},after=Struct{id=1003,name=wangwu,sex=male},source=Struct{version=1.5.2.Final,connector=mysql,name=mysql_binlog_source,ts_ms=1673531342000,db=cdc_test,table=user_info,server_id=1,file=mysql-bin.000001,pos=943,row=0},op=u,ts_ms=1673531338605}, valueSchema=Schema{mysql_binlog_source.cdc_test.user_info.Envelope:STRUCT}, timestamp=null, headers=ConnectHeaders(headers=)}
* */
@Slf4j
public class CustomerDeserializationSchema implements DebeziumDeserializationSchema<String> {



    /*TODO 明确自己想要的数据格式
     * (
     *   "db":"",
     *   "tableName":"",
     *   "before":{"id":"1001","name":""...},
     *   "after":{"id":"1001","name":""...},
     *   "op":""
     * )
     * */

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) throws Exception {
        //创建JSON 对象用于封装结果数据
        JSONObject result = new JSONObject();

        //获取库名&表名
        String topic = sourceRecord.topic();//根据上面样式通过sourceRecord.键值，获取对应值
        //获取结果：topic='mysql_binlog_source.cdc_test.user_info'
        String[] fields = topic.split("\\.");//安装`.`进行分割（.需要转义）

        //添加对应的库名和表名键值对
        result.put("db", fields[1]);
        result.put("tableName", fields[2]);

        //获取before数据
        Struct value = (Struct) sourceRecord.value();//需要进行强转下,注意导的是：org.apache.kafka.connect.data.Struct 这个包
        Struct before = value.getStruct("before");//通过指定before，获取before字段的数据
        JSONObject beforeJson = new JSONObject();
        if (before != null) {// before字段是有可能为空的(比如读取[op=r]、插入操作[op=c]...)，所以要进行判断
            Schema schema = before.schema();//获取before 的schema信息
            List<Field> fieldList = schema.fields();//获取before里的全部字段

            for (Field field : fieldList) {
                // 通过field.name()获取对应的字段名,before.get(field)根据字段名,获取对应的值
                beforeJson.put(field.name(), before.get(field));
            }
        }
        result.put("before", beforeJson);//把before信息添加进去

        //同理获取after
        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            Schema schema = after.schema();
            List<Field> fieldList = schema.fields();

            for (Field field : fieldList) {
                afterJson.put(field.name(), after.get(field));
            }
        }
        result.put("after", afterJson);

        //获取操作类型(operation不能直接通过sourceRecord获取)
        Envelope.Operation operation = Envelope.operationFor(sourceRecord);//注意导包：io.debezium.data.Envelope
        //将operation信息添加进去
        result.put("op", operation);
//        CdcDataProcessCache(result);
        //输出数据
        collector.collect(result.toJSONString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        // 返回类型
        return BasicTypeInfo.STRING_TYPE_INFO;
    }


    private String conversionTime(String timeStamp) {
        //yyyy-MM-dd HH:mm:ss 转换的时间格式  可以自定义
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        //转换
        return sdf.format(new Date(Long.parseLong(timeStamp)));
    }



    /**
     * 临时写入缓存
     */
    private void CdcDataProcessCache(JSONObject result) {


//        //读取Springboot配置文件
//        ApplicationContext ctx = new ClassPathXmlApplicationContext("applicationContext.xml");
//        //通过类型来获取类
//        RedisUtil redisUtils = (RedisUtil) ctx.getBean(RedisUtil.class);


        if (result.getString("db").equals("study")) {
            if (result.getString("tableName").equals("token_user")) {
                switch (result.getString("op")) {
                    case "UPDATE":
//                        log.info("xxxxxxxxxx"+redisUtil.get("categroy::findTokenByUserName").toString());
                        break;
                    case "CREATE":

                        break;
                    case "DELETE":

                        break;
                }
            }
        }

    }
}
